# @Time : 2024/6/13 14:50
# @Author : ZHUYI
# @File : producer_juejin
import spider_juejin
from kafka import KafkaProducer
import json


def producer_juejin(topic):
    # 连接kafka
    producer = KafkaProducer(bootstrap_servers='localhost:9092',
                             value_serializer=lambda v: json.dumps(v).encode('utf-8'))  # 连接kafka
    # 调用爬虫函数，获取掘金平台的数据
    result = spider_juejin.spiderAndSave()
    print("生产者启动。。。。")
    # 将爬虫获取的数据发送到相应的的Kafka主题
    producer.send(topic, result)  # 发送的topic为json
    # 确保所有发送的消息都被发送到服务器
    producer.flush()
    producer.close()
    print("生产者结束。。。。")


